package com.hu.flink12.api.transformation;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author: hujianjun
 * @Date: 2021/2/4 0:17
 * @Describe:
 */
public class OutputTagStream {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        OutputTag<Long> oddTag = new OutputTag<>("odd", TypeInformation.of(Long.class));
        OutputTag<Long> evenTag = new OutputTag<>("even", TypeInformation.of(Long.class));

        //TODO 2.读取数据源
        DataStream<Long> ds = env.fromSequence(1, 10);

        //TODO 3.转换操作
        SingleOutputStreamOperator<Long> process = ds.process(new ProcessFunction<Long, Long>() {
            @Override
            public void processElement(Long aLong, Context ctx, Collector<Long> collector) throws Exception {
                if (aLong % 2 == 0) {
                    ctx.output(evenTag, aLong);
                } else {
                    ctx.output(oddTag, aLong);
                }
            }
        });


        //TODO 4.sink操作
        process.getSideOutput(oddTag).print("odd");
        process.getSideOutput(evenTag).print("even");

        //TODO 5.执行程序
        env.execute();
    }
}
